diff --git a/swh/web/browse/utils.py b/swh/web/browse/utils.py
index ebb81037..2f924f4f 100644
--- a/swh/web/browse/utils.py
+++ b/swh/web/browse/utils.py
@@ -1,1100 +1,1100 @@
# Copyright (C) 2017-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import base64
import magic
import pypandoc
import stat
import textwrap
from collections import defaultdict
from threading import Lock
from django.core.cache import cache
from django.utils.safestring import mark_safe
from django.utils.html import escape
from swh.model.identifiers import persistent_identifier
from swh.web.common import highlightjs, service
from swh.web.common.exc import NotFoundExc, http_status_code_message
from swh.web.common.origin_visits import get_origin_visit
from swh.web.common.utils import (
reverse, format_utc_iso_date, get_swh_persistent_id,
swh_object_icons
)
from swh.web.config import get_config
def get_directory_entries(sha1_git):
"""Function that retrieves the content of a directory
from the archive.
The directories entries are first sorted in lexicographical order.
Sub-directories and regular files are then extracted.
Args:
sha1_git: sha1_git identifier of the directory
Returns:
A tuple whose first member corresponds to the sub-directories list
and second member the regular files list
Raises:
NotFoundExc if the directory is not found
"""
cache_entry_id = 'directory_entries_%s' % sha1_git
cache_entry = cache.get(cache_entry_id)
if cache_entry:
return cache_entry
entries = list(service.lookup_directory(sha1_git))
for e in entries:
e['perms'] = stat.filemode(e['perms'])
if e['type'] == 'rev':
# modify dir entry name to explicitly show it points
# to a revision
e['name'] = '%s @ %s' % (e['name'], e['target'][:7])
dirs = [e for e in entries if e['type'] in ('dir', 'rev')]
files = [e for e in entries if e['type'] == 'file']
dirs = sorted(dirs, key=lambda d: d['name'])
files = sorted(files, key=lambda f: f['name'])
cache.set(cache_entry_id, (dirs, files))
return dirs, files
_lock = Lock()
def get_mimetype_and_encoding_for_content(content):
"""Function that returns the mime type and the encoding associated to
a content buffer using the magic module under the hood.
Args:
content (bytes): a content buffer
Returns:
A tuple (mimetype, encoding), for instance ('text/plain', 'us-ascii'),
associated to the provided content.
"""
# https://pypi.org/project/python-magic/
# packaged as python3-magic in debian buster
if hasattr(magic, 'from_buffer'):
m = magic.Magic(mime=True, mime_encoding=True)
mime_encoding = m.from_buffer(content)
mime_type, encoding = mime_encoding.split(';')
encoding = encoding.replace(' charset=', '')
# https://pypi.org/project/file-magic/
# packaged as python3-magic in debian stretch
else:
# TODO: Remove that code when production environment is upgraded
# to debian buster
# calls to the file-magic API are not thread-safe so they must
# be protected with a Lock to guarantee they will succeed
_lock.acquire()
magic_result = magic.detect_from_content(content)
_lock.release()
mime_type = magic_result.mime_type
encoding = magic_result.encoding
return mime_type, encoding
# maximum authorized content size in bytes for HTML display
# with code highlighting
content_display_max_size = get_config()['content_display_max_size']
snapshot_content_max_size = get_config()['snapshot_content_max_size']
def _re_encode_content(mimetype, encoding, content_data):
# encode textual content to utf-8 if needed
if mimetype.startswith('text/'):
# probably a malformed UTF-8 content, re-encode it
# by replacing invalid chars with a substitution one
if encoding == 'unknown-8bit':
content_data = content_data.decode('utf-8', 'replace')\
.encode('utf-8')
elif encoding not in ['utf-8', 'binary']:
content_data = content_data.decode(encoding, 'replace')\
.encode('utf-8')
elif mimetype.startswith('application/octet-stream'):
# file may detect a text content as binary
# so try to decode it for display
encodings = ['us-ascii', 'utf-8']
encodings += ['iso-8859-%s' % i for i in range(1, 17)]
for enc in encodings:
try:
content_data = content_data.decode(enc).encode('utf-8')
except Exception:
pass
else:
# ensure display in content view
encoding = enc
mimetype = 'text/plain'
break
return mimetype, encoding, content_data
def request_content(query_string, max_size=content_display_max_size,
raise_if_unavailable=True, re_encode=True):
"""Function that retrieves a content from the archive.
Raw bytes content is first retrieved, then the content mime type.
If the mime type is not stored in the archive, it will be computed
using Python magic module.
Args:
query_string: a string of the form "[ALGO_HASH:]HASH" where
optional ALGO_HASH can be either ``sha1``, ``sha1_git``,
``sha256``, or ``blake2s256`` (default to ``sha1``) and HASH
the hexadecimal representation of the hash value
max_size: the maximum size for a content to retrieve (default to 1MB,
no size limit if None)
Returns:
A tuple whose first member corresponds to the content raw bytes
and second member the content mime type
Raises:
NotFoundExc if the content is not found
"""
content_data = service.lookup_content(query_string)
filetype = None
language = None
license = None
# requests to the indexer db may fail so properly handle
# those cases in order to avoid content display errors
try:
filetype = service.lookup_content_filetype(query_string)
language = service.lookup_content_language(query_string)
license = service.lookup_content_license(query_string)
except Exception:
pass
mimetype = 'unknown'
encoding = 'unknown'
if filetype:
mimetype = filetype['mimetype']
encoding = filetype['encoding']
# workaround when encountering corrupted data due to implicit
# conversion from bytea to text in the indexer db (see T818)
# TODO: Remove that code when all data have been correctly converted
if mimetype.startswith('\\'):
filetype = None
content_data['error_code'] = 200
content_data['error_message'] = ''
content_data['error_description'] = ''
if not max_size or content_data['length'] < max_size:
try:
content_raw = service.lookup_content_raw(query_string)
except Exception as e:
if raise_if_unavailable:
raise e
else:
content_data['raw_data'] = None
content_data['error_code'] = 404
content_data['error_description'] = \
'The bytes of the content are currently not available in the archive.' # noqa
content_data['error_message'] = \
http_status_code_message[content_data['error_code']]
else:
content_data['raw_data'] = content_raw['data']
if not filetype:
mimetype, encoding = \
get_mimetype_and_encoding_for_content(content_data['raw_data']) # noqa
if re_encode:
mimetype, encoding, raw_data = _re_encode_content(
mimetype, encoding, content_data['raw_data'])
content_data['raw_data'] = raw_data
else:
content_data['raw_data'] = None
content_data['mimetype'] = mimetype
content_data['encoding'] = encoding
if language:
content_data['language'] = language['lang']
else:
content_data['language'] = 'not detected'
if license:
content_data['licenses'] = ', '.join(license['facts'][0]['licenses'])
else:
content_data['licenses'] = 'not detected'
return content_data
_browsers_supported_image_mimes = set(['image/gif', 'image/png',
'image/jpeg', 'image/bmp',
'image/webp', 'image/svg',
'image/svg+xml'])
def prepare_content_for_display(content_data, mime_type, path):
"""Function that prepares a content for HTML display.
The function tries to associate a programming language to a
content in order to perform syntax highlighting client-side
using highlightjs. The language is determined using either
the content filename or its mime type.
If the mime type corresponds to an image format supported
by web browsers, the content will be encoded in base64
for displaying the image.
Args:
content_data (bytes): raw bytes of the content
mime_type (string): mime type of the content
path (string): path of the content including filename
Returns:
A dict containing the content bytes (possibly different from the one
provided as parameter if it is an image) under the key 'content_data
and the corresponding highlightjs language class under the
key 'language'.
"""
language = highlightjs.get_hljs_language_from_filename(path)
if not language:
language = highlightjs.get_hljs_language_from_mime_type(mime_type)
if not language:
language = 'nohighlight'
elif mime_type.startswith('application/'):
mime_type = mime_type.replace('application/', 'text/')
if mime_type.startswith('image/'):
if mime_type in _browsers_supported_image_mimes:
content_data = base64.b64encode(content_data)
content_data = content_data.decode('utf-8')
else:
content_data = None
if mime_type.startswith('image/svg'):
mime_type = 'image/svg+xml'
return {'content_data': content_data,
'language': language,
'mimetype': mime_type}
def process_snapshot_branches(snapshot):
"""
Process a dictionary describing snapshot branches: extract those
targeting revisions and releases, put them in two different lists,
then sort those lists in lexicographical order of the branches' names.
Args:
snapshot_branches (dict): A dict describing the branches of a snapshot
as returned for instance by
:func:`swh.web.common.service.lookup_snapshot`
Returns:
tuple: A tuple whose first member is the sorted list of branches
targeting revisions and second member the sorted list of branches
targeting releases
"""
snapshot_branches = snapshot['branches']
branches = {}
branch_aliases = {}
releases = {}
revision_to_branch = defaultdict(set)
revision_to_release = defaultdict(set)
release_to_branch = defaultdict(set)
for branch_name, target in snapshot_branches.items():
if not target:
# FIXME: display branches with an unknown target anyway
continue
target_id = target['target']
target_type = target['target_type']
if target_type == 'revision':
branches[branch_name] = {
'name': branch_name,
'revision': target_id,
}
revision_to_branch[target_id].add(branch_name)
elif target_type == 'release':
release_to_branch[target_id].add(branch_name)
elif target_type == 'alias':
branch_aliases[branch_name] = target_id
# FIXME: handle pointers to other object types
def _enrich_release_branch(branch, release):
releases[branch] = {
'name': release['name'],
'branch_name': branch,
'date': format_utc_iso_date(release['date']),
'id': release['id'],
'message': release['message'],
'target_type': release['target_type'],
'target': release['target'],
}
def _enrich_revision_branch(branch, revision):
branches[branch].update({
'revision': revision['id'],
'directory': revision['directory'],
'date': format_utc_iso_date(revision['date']),
'message': revision['message']
})
releases_info = service.lookup_release_multiple(
release_to_branch.keys()
)
for release in releases_info:
branches_to_update = release_to_branch[release['id']]
for branch in branches_to_update:
_enrich_release_branch(branch, release)
if release['target_type'] == 'revision':
revision_to_release[release['target']].update(
branches_to_update
)
revisions = service.lookup_revision_multiple(
set(revision_to_branch.keys()) | set(revision_to_release.keys())
)
for revision in revisions:
if not revision:
continue
for branch in revision_to_branch[revision['id']]:
_enrich_revision_branch(branch, revision)
for release in revision_to_release[revision['id']]:
releases[release]['directory'] = revision['directory']
for branch_alias, branch_target in branch_aliases.items():
if branch_target in branches:
branches[branch_alias] = dict(branches[branch_target])
else:
snp = service.lookup_snapshot(snapshot['id'],
branches_from=branch_target,
branches_count=1)
if snp and branch_target in snp['branches']:
if snp['branches'][branch_target] is None:
continue
target_type = snp['branches'][branch_target]['target_type']
target = snp['branches'][branch_target]['target']
if target_type == 'revision':
branches[branch_alias] = snp['branches'][branch_target]
revision = service.lookup_revision(target)
_enrich_revision_branch(branch_alias, revision)
elif target_type == 'release':
release = service.lookup_release(target)
_enrich_release_branch(branch_alias, release)
if branch_alias in branches:
branches[branch_alias]['name'] = branch_alias
ret_branches = list(sorted(branches.values(), key=lambda b: b['name']))
ret_releases = list(sorted(releases.values(), key=lambda b: b['name']))
return ret_branches, ret_releases
def get_snapshot_content(snapshot_id):
"""Returns the lists of branches and releases
associated to a swh snapshot.
That list is put in cache in order to speedup the navigation
in the swh-web/browse ui.
.. warning:: At most 1000 branches contained in the snapshot
will be returned for performance reasons.
Args:
snapshot_id (str): hexadecimal representation of the snapshot
identifier
Returns:
A tuple with two members. The first one is a list of dict describing
the snapshot branches. The second one is a list of dict describing the
snapshot releases.
Raises:
NotFoundExc if the snapshot does not exist
"""
cache_entry_id = 'swh_snapshot_%s' % snapshot_id
cache_entry = cache.get(cache_entry_id)
if cache_entry:
return cache_entry['branches'], cache_entry['releases']
branches = []
releases = []
if snapshot_id:
snapshot = service.lookup_snapshot(
snapshot_id, branches_count=snapshot_content_max_size)
branches, releases = process_snapshot_branches(snapshot)
cache.set(cache_entry_id, {
'branches': branches,
'releases': releases,
})
return branches, releases
def get_origin_visit_snapshot(origin_info, visit_ts=None, visit_id=None,
snapshot_id=None):
"""Returns the lists of branches and releases
associated to a swh origin for a given visit.
The visit is expressed by a timestamp. In the latter case,
the closest visit from the provided timestamp will be used.
If no visit parameter is provided, it returns the list of branches
found for the latest visit.
That list is put in cache in order to speedup the navigation
in the swh-web/browse ui.
.. warning:: At most 1000 branches contained in the snapshot
will be returned for performance reasons.
Args:
origin_info (dict): a dict filled with origin information
(id, url, type)
visit_ts (int or str): an ISO date string or Unix timestamp to parse
visit_id (int): optional visit id for disambiguation in case
several visits have the same timestamp
Returns:
A tuple with two members. The first one is a list of dict describing
the origin branches for the given visit.
The second one is a list of dict describing the origin releases
for the given visit.
Raises:
NotFoundExc if the origin or its visit are not found
"""
visit_info = get_origin_visit(origin_info, visit_ts, visit_id, snapshot_id)
return get_snapshot_content(visit_info['snapshot'])
def gen_link(url, link_text=None, link_attrs=None):
"""
Utility function for generating an HTML link to insert
in Django templates.
Args:
url (str): an url
link_text (str): optional text for the produced link,
if not provided the url will be used
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form 'link_text'
"""
attrs = ' '
if link_attrs:
for k, v in link_attrs.items():
attrs += '%s="%s" ' % (k, v)
if not link_text:
link_text = url
link = '%s' \
% (attrs, escape(url), escape(link_text))
return mark_safe(link)
def _snapshot_context_query_params(snapshot_context):
query_params = None
if snapshot_context and snapshot_context['origin_info']:
origin_info = snapshot_context['origin_info']
query_params = {'origin': origin_info['url']}
if 'timestamp' in snapshot_context['url_args']:
query_params['timestamp'] = \
snapshot_context['url_args']['timestamp']
if 'visit_id' in snapshot_context['query_params']:
query_params['visit_id'] = \
snapshot_context['query_params']['visit_id']
elif snapshot_context:
query_params = {'snapshot_id': snapshot_context['snapshot_id']}
return query_params
def gen_revision_url(revision_id, snapshot_context=None):
"""
Utility function for generating an url to a revision.
Args:
revision_id (str): a revision id
snapshot_context (dict): if provided, generate snapshot-dependent
browsing url
Returns:
str: The url to browse the revision
"""
query_params = _snapshot_context_query_params(snapshot_context)
return reverse('browse-revision',
url_args={'sha1_git': revision_id},
query_params=query_params)
def gen_revision_link(revision_id, shorten_id=False, snapshot_context=None,
link_text='Browse',
link_attrs={'class': 'btn btn-default btn-sm',
'role': 'button'}):
"""
Utility function for generating a link to a revision HTML view
to insert in Django templates.
Args:
revision_id (str): a revision id
shorten_id (boolean): whether to shorten the revision id to 7
characters for the link text
snapshot_context (dict): if provided, generate snapshot-dependent
browsing link
link_text (str): optional text for the generated link
(the revision id will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
str: An HTML link in the form 'revision_id'
"""
if not revision_id:
return None
revision_url = gen_revision_url(revision_id, snapshot_context)
if shorten_id:
return gen_link(revision_url, revision_id[:7], link_attrs)
else:
if not link_text:
link_text = revision_id
return gen_link(revision_url, link_text, link_attrs)
def gen_directory_link(sha1_git, snapshot_context=None, link_text='Browse',
link_attrs={'class': 'btn btn-default btn-sm',
'role': 'button'}):
"""
Utility function for generating a link to a directory HTML view
to insert in Django templates.
Args:
sha1_git (str): directory identifier
link_text (str): optional text for the generated link
(the directory id will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form 'link_text'
"""
if not sha1_git:
return None
query_params = _snapshot_context_query_params(snapshot_context)
directory_url = reverse('browse-directory',
url_args={'sha1_git': sha1_git},
query_params=query_params)
if not link_text:
link_text = sha1_git
return gen_link(directory_url, link_text, link_attrs)
def gen_snapshot_link(snapshot_id, snapshot_context=None, link_text='Browse',
link_attrs={'class': 'btn btn-default btn-sm',
'role': 'button'}):
"""
Utility function for generating a link to a snapshot HTML view
to insert in Django templates.
Args:
snapshot_id (str): snapshot identifier
link_text (str): optional text for the generated link
(the snapshot id will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form 'link_text'
"""
query_params = _snapshot_context_query_params(snapshot_context)
snapshot_url = reverse('browse-snapshot',
url_args={'snapshot_id': snapshot_id},
query_params=query_params)
if not link_text:
link_text = snapshot_id
return gen_link(snapshot_url, link_text, link_attrs)
def gen_content_link(sha1_git, snapshot_context=None, link_text='Browse',
link_attrs={'class': 'btn btn-default btn-sm',
'role': 'button'}):
"""
Utility function for generating a link to a content HTML view
to insert in Django templates.
Args:
sha1_git (str): content identifier
link_text (str): optional text for the generated link
(the content sha1_git will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form 'link_text'
"""
if not sha1_git:
return None
query_params = _snapshot_context_query_params(snapshot_context)
content_url = reverse('browse-content',
url_args={'query_string': 'sha1_git:' + sha1_git},
query_params=query_params)
if not link_text:
link_text = sha1_git
return gen_link(content_url, link_text, link_attrs)
def get_revision_log_url(revision_id, snapshot_context=None):
"""
Utility function for getting the URL for a revision log HTML view
(possibly in the context of an origin).
Args:
revision_id (str): revision identifier the history heads to
snapshot_context (dict): if provided, generate snapshot-dependent
browsing link
Returns:
The revision log view URL
"""
query_params = {'revision': revision_id}
if snapshot_context and snapshot_context['origin_info']:
origin_info = snapshot_context['origin_info']
url_args = {'origin_url': origin_info['url']}
if 'timestamp' in snapshot_context['url_args']:
url_args['timestamp'] = \
snapshot_context['url_args']['timestamp']
if 'visit_id' in snapshot_context['query_params']:
query_params['visit_id'] = \
snapshot_context['query_params']['visit_id']
revision_log_url = reverse('browse-origin-log',
url_args=url_args,
query_params=query_params)
elif snapshot_context:
url_args = {'snapshot_id': snapshot_context['snapshot_id']}
revision_log_url = reverse('browse-snapshot-log',
url_args=url_args,
query_params=query_params)
else:
revision_log_url = reverse('browse-revision-log',
url_args={'sha1_git': revision_id})
return revision_log_url
def gen_revision_log_link(revision_id, snapshot_context=None,
link_text='Browse',
link_attrs={'class': 'btn btn-default btn-sm',
'role': 'button'}):
"""
Utility function for generating a link to a revision log HTML view
(possibly in the context of an origin) to insert in Django templates.
Args:
revision_id (str): revision identifier the history heads to
snapshot_context (dict): if provided, generate snapshot-dependent
browsing link
link_text (str): optional text to use for the generated link
(the revision id will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form
'link_text'
"""
if not revision_id:
return None
revision_log_url = get_revision_log_url(revision_id, snapshot_context)
if not link_text:
link_text = revision_id
return gen_link(revision_log_url, link_text, link_attrs)
def gen_person_mail_link(person, link_text=None):
"""
Utility function for generating a mail link to a person to insert
in Django templates.
Args:
person (dict): dictionary containing person data
(*name*, *email*, *fullname*)
link_text (str): optional text to use for the generated mail link
(the person name will be used by default)
Returns:
str: A mail link to the person or the person name if no email is
present in person data
"""
person_name = person['name'] or person['fullname'] or 'None'
if link_text is None:
link_text = person_name
person_email = person['email'] if person['email'] else None
if person_email is None and '@' in person_name and ' ' not in person_name:
person_email = person_name
if person_email:
return gen_link(url='mailto:%s' % person_email,
link_text=link_text)
else:
return person_name
def gen_release_link(sha1_git, snapshot_context=None, link_text='Browse',
link_attrs={'class': 'btn btn-default btn-sm',
'role': 'button'}):
"""
Utility function for generating a link to a release HTML view
to insert in Django templates.
Args:
sha1_git (str): release identifier
link_text (str): optional text for the generated link
(the release id will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form 'link_text'
"""
query_params = _snapshot_context_query_params(snapshot_context)
release_url = reverse('browse-release',
url_args={'sha1_git': sha1_git},
query_params=query_params)
if not link_text:
link_text = sha1_git
return gen_link(release_url, link_text, link_attrs)
def format_log_entries(revision_log, per_page, snapshot_context=None):
"""
Utility functions that process raw revision log data for HTML display.
Its purpose is to:
* add links to relevant browse views
* format date in human readable format
* truncate the message log
Args:
revision_log (list): raw revision log as returned by the swh-web api
per_page (int): number of log entries per page
snapshot_context (dict): if provided, generate snapshot-dependent
browsing link
"""
revision_log_data = []
for i, rev in enumerate(revision_log):
if i == per_page:
break
author_name = 'None'
author_fullname = 'None'
committer_fullname = 'None'
if rev['author']:
author_name = gen_person_mail_link(rev['author'])
author_fullname = rev['author']['fullname']
if rev['committer']:
committer_fullname = rev['committer']['fullname']
author_date = format_utc_iso_date(rev['date'])
committer_date = format_utc_iso_date(rev['committer_date'])
tooltip = 'revision %s\n' % rev['id']
tooltip += 'author: %s\n' % author_fullname
tooltip += 'author date: %s\n' % author_date
tooltip += 'committer: %s\n' % committer_fullname
tooltip += 'committer date: %s\n\n' % committer_date
if rev['message']:
tooltip += textwrap.indent(rev['message'], ' '*4)
revision_log_data.append({
'author': author_name,
'id': rev['id'][:7],
'message': rev['message'],
'date': author_date,
'commit_date': committer_date,
'url': gen_revision_url(rev['id'], snapshot_context),
'tooltip': tooltip
})
return revision_log_data
def get_snapshot_context(snapshot_id=None, origin_url=None,
timestamp=None, visit_id=None):
"""
Utility function to compute relevant information when navigating
the archive in a snapshot context. The snapshot is either
referenced by its id or it will be retrieved from an origin visit.
Args:
snapshot_id (str): hexadecimal representation of a snapshot identifier,
all other parameters will be ignored if it is provided
origin_url (str): the origin_url
(e.g. https://github.com/(user)/(repo)/)
timestamp (str): a datetime string for retrieving the closest
visit of the origin
visit_id (int): optional visit id for disambiguation in case
of several visits with the same timestamp
Returns:
A dict with the following entries:
* origin_info: dict containing origin information
* visit_info: dict containing visit information
* branches: the list of branches for the origin found
during the visit
* releases: the list of releases for the origin found
during the visit
* origin_browse_url: the url to browse the origin
* origin_branches_url: the url to browse the origin branches
* origin_releases_url': the url to browse the origin releases
* origin_visit_url: the url to browse the snapshot of the origin
found during the visit
* url_args: dict containing url arguments to use when browsing in
the context of the origin and its visit
Raises:
NotFoundExc: if no snapshot is found for the visit of an origin.
"""
origin_info = None
visit_info = None
url_args = None
query_params = {}
branches = []
releases = []
browse_url = None
visit_url = None
branches_url = None
releases_url = None
swh_type = 'snapshot'
if origin_url:
swh_type = 'origin'
origin_info = service.lookup_origin({'url': origin_url})
visit_info = get_origin_visit(origin_info, timestamp, visit_id,
snapshot_id)
fmt_date = format_utc_iso_date(visit_info['date'])
visit_info['fmt_date'] = fmt_date
snapshot_id = visit_info['snapshot']
if not snapshot_id:
raise NotFoundExc('No snapshot associated to the visit of origin '
'%s on %s' % (escape(origin_url), fmt_date))
# provided timestamp is not necessarily equals to the one
# of the retrieved visit, so get the exact one in order
# use it in the urls generated below
if timestamp:
timestamp = visit_info['date']
branches, releases = \
get_origin_visit_snapshot(origin_info, timestamp, visit_id,
snapshot_id)
url_args = {'origin_url': origin_info['url']}
query_params = {'visit_id': visit_id}
browse_url = reverse('browse-origin-visits',
url_args=url_args)
if timestamp:
url_args['timestamp'] = format_utc_iso_date(timestamp,
'%Y-%m-%dT%H:%M:%S')
visit_url = reverse('browse-origin-directory',
url_args=url_args,
query_params=query_params)
visit_info['url'] = visit_url
branches_url = reverse('browse-origin-branches',
url_args=url_args,
query_params=query_params)
releases_url = reverse('browse-origin-releases',
url_args=url_args,
query_params=query_params)
elif snapshot_id:
branches, releases = get_snapshot_content(snapshot_id)
url_args = {'snapshot_id': snapshot_id}
browse_url = reverse('browse-snapshot',
url_args=url_args)
branches_url = reverse('browse-snapshot-branches',
url_args=url_args)
releases_url = reverse('browse-snapshot-releases',
url_args=url_args)
releases = list(reversed(releases))
- snapshot_size = service.lookup_snapshot_size(snapshot_id)
+ snapshot_sizes = service.lookup_snapshot_sizes(snapshot_id)
- is_empty = sum(snapshot_size.values()) == 0
+ is_empty = sum(snapshot_sizes.values()) == 0
swh_snp_id = persistent_identifier('snapshot', snapshot_id)
return {
'swh_type': swh_type,
'swh_object_id': swh_snp_id,
'snapshot_id': snapshot_id,
- 'snapshot_size': snapshot_size,
+ 'snapshot_sizes': snapshot_sizes,
'is_empty': is_empty,
'origin_info': origin_info,
'visit_info': visit_info,
'branches': branches,
'releases': releases,
'branch': None,
'release': None,
'browse_url': browse_url,
'branches_url': branches_url,
'releases_url': releases_url,
'url_args': url_args,
'query_params': query_params
}
# list of common readme names ordered by preference
# (lower indices have higher priority)
_common_readme_names = [
"readme.markdown",
"readme.md",
"readme.rst",
"readme.txt",
"readme"
]
def get_readme_to_display(readmes):
"""
Process a list of readme files found in a directory
in order to find the adequate one to display.
Args:
readmes: a list of dict where keys are readme file names and values
are readme sha1s
Returns:
A tuple (readme_name, readme_sha1)
"""
readme_name = None
readme_url = None
readme_sha1 = None
readme_html = None
lc_readmes = {k.lower(): {'orig_name': k, 'sha1': v}
for k, v in readmes.items()}
# look for readme names according to the preference order
# defined by the _common_readme_names list
for common_readme_name in _common_readme_names:
if common_readme_name in lc_readmes:
readme_name = lc_readmes[common_readme_name]['orig_name']
readme_sha1 = lc_readmes[common_readme_name]['sha1']
readme_url = reverse('browse-content-raw',
url_args={'query_string': readme_sha1},
query_params={'re_encode': 'true'})
break
# otherwise pick the first readme like file if any
if not readme_name and len(readmes.items()) > 0:
readme_name = next(iter(readmes))
readme_sha1 = readmes[readme_name]
readme_url = reverse('browse-content-raw',
url_args={'query_string': readme_sha1},
query_params={'re_encode': 'true'})
# convert rst README to html server side as there is
# no viable solution to perform that task client side
if readme_name and readme_name.endswith('.rst'):
cache_entry_id = 'readme_%s' % readme_sha1
cache_entry = cache.get(cache_entry_id)
if cache_entry:
readme_html = cache_entry
else:
try:
rst_doc = request_content(readme_sha1)
readme_html = pypandoc.convert_text(rst_doc['raw_data'],
'html', format='rst')
cache.set(cache_entry_id, readme_html)
except Exception:
readme_html = 'Readme bytes are not available'
return readme_name, readme_url, readme_html
def get_swh_persistent_ids(swh_objects, snapshot_context=None):
"""
Returns a list of dict containing info related to persistent
identifiers of swh objects.
Args:
swh_objects (list): a list of dict with the following keys:
* type: swh object type
(content/directory/release/revision/snapshot)
* id: swh object id
snapshot_context (dict): optional parameter describing the snapshot in
which the object has been found
Returns:
list: a list of dict with the following keys:
* object_type: the swh object type
(content/directory/release/revision/snapshot)
* object_icon: the swh object icon to use in HTML views
* swh_id: the computed swh object persistent identifier
* swh_id_url: the url resolving the persistent identifier
* show_options: boolean indicating if the persistent id options
must be displayed in persistent ids HTML view
"""
swh_ids = []
for swh_object in swh_objects:
if not swh_object['id']:
continue
swh_id = get_swh_persistent_id(swh_object['type'], swh_object['id'])
show_options = swh_object['type'] == 'content' or \
(snapshot_context and snapshot_context['origin_info'] is not None)
object_icon = swh_object_icons[swh_object['type']]
swh_ids.append({
'object_type': swh_object['type'],
'object_icon': object_icon,
'swh_id': swh_id,
'swh_id_url': reverse('browse-swh-id',
url_args={'swh_id': swh_id}),
'show_options': show_options
})
return swh_ids
diff --git a/swh/web/browse/views/utils/snapshot_context.py b/swh/web/browse/views/utils/snapshot_context.py
index d1c814ce..4310da62 100644
--- a/swh/web/browse/views/utils/snapshot_context.py
+++ b/swh/web/browse/views/utils/snapshot_context.py
@@ -1,925 +1,925 @@
# Copyright (C) 2018-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
# Utility module implementing Django views for browsing the archive
# in a snapshot context.
# Its purpose is to factorize code for the views reachable from the
# /origin/.* and /snapshot/.* endpoints.
from django.shortcuts import render
from django.template.defaultfilters import filesizeformat
from django.utils.html import escape
from swh.model.identifiers import snapshot_identifier
from swh.web.browse.utils import (
get_snapshot_context, get_directory_entries, gen_directory_link,
gen_revision_link, request_content, gen_content_link,
prepare_content_for_display, content_display_max_size,
format_log_entries, gen_revision_log_link, gen_release_link,
get_readme_to_display, get_swh_persistent_ids,
gen_snapshot_link, process_snapshot_branches
)
from swh.web.common import service, highlightjs
from swh.web.common.exc import (
handle_view_exception, NotFoundExc
)
from swh.web.common.utils import (
reverse, gen_path_info, format_utc_iso_date, swh_object_icons
)
_empty_snapshot_id = snapshot_identifier({'branches': {}})
def _get_branch(branches, branch_name, snapshot_id):
"""
Utility function to get a specific branch from a branches list.
Its purpose is to get the default HEAD branch as some software origin
(e.g those with svn type) does not have it. In that latter case, check
if there is a master branch instead and returns it.
"""
filtered_branches = [b for b in branches if b['name'] == branch_name]
if filtered_branches:
return filtered_branches[0]
elif branch_name == 'HEAD':
filtered_branches = [b for b in branches
if b['name'].endswith('master')]
if filtered_branches:
return filtered_branches[0]
elif branches:
return branches[0]
else:
# case where a large branches list has been truncated
snp = service.lookup_snapshot(snapshot_id,
branches_from=branch_name,
branches_count=1,
target_types=['revision', 'alias'])
snp_branch, _ = process_snapshot_branches(snp)
if snp_branch and snp_branch[0]['name'] == branch_name:
branches.append(snp_branch[0])
return snp_branch[0]
def _get_release(releases, release_name, snapshot_id):
"""
Utility function to get a specific release from a releases list.
Returns None if the release can not be found in the list.
"""
filtered_releases = [r for r in releases if r['name'] == release_name]
if filtered_releases:
return filtered_releases[0]
else:
# case where a large branches list has been truncated
for branch_name in (release_name, f'refs/tags/{release_name}'):
snp = service.lookup_snapshot(snapshot_id,
branches_from=branch_name,
branches_count=1,
target_types=['release'])
_, snp_release = process_snapshot_branches(snp)
if snp_release and snp_release[0]['name'] == release_name:
releases.append(snp_release[0])
return snp_release[0]
-def _branch_not_found(branch_type, branch, snapshot_id, snapshot_size,
+def _branch_not_found(branch_type, branch, snapshot_id, snapshot_sizes,
origin_info, timestamp, visit_id):
"""
Utility function to raise an exception when a specified branch/release
can not be found.
"""
if branch_type == 'branch':
branch_type = 'Branch'
branch_type_plural = 'branches'
target_type = 'revision'
else:
branch_type = 'Release'
branch_type_plural = 'releases'
target_type = 'release'
- if snapshot_id and snapshot_size[target_type] == 0:
+ if snapshot_id and snapshot_sizes[target_type] == 0:
msg = ('Snapshot with id %s has an empty list'
' of %s!' % (snapshot_id, branch_type_plural))
elif snapshot_id:
msg = ('%s %s for snapshot with id %s'
' not found!' % (branch_type, branch, snapshot_id))
- elif visit_id and snapshot_size[target_type] == 0:
+ elif visit_id and snapshot_sizes[target_type] == 0:
msg = ('Origin with url %s'
' for visit with id %s has an empty list'
' of %s!' % (origin_info['url'], visit_id,
branch_type_plural))
elif visit_id:
msg = ('%s %s associated to visit with'
' id %s for origin with url %s'
' not found!' % (branch_type, branch, visit_id,
origin_info['url']))
- elif snapshot_size[target_type] == 0:
+ elif snapshot_sizes[target_type] == 0:
msg = ('Origin with url %s'
' for visit with timestamp %s has an empty list'
' of %s!' % (origin_info['url'],
timestamp, branch_type_plural))
else:
msg = ('%s %s associated to visit with'
' timestamp %s for origin with '
'url %s not found!' % (branch_type, branch, timestamp,
origin_info['url']))
raise NotFoundExc(escape(msg))
def _process_snapshot_request(request, snapshot_id=None,
origin_url=None, timestamp=None, path=None,
browse_context='directory'):
"""
Utility function to perform common input request processing
for snapshot context views.
"""
visit_id = request.GET.get('visit_id', None)
snapshot_context = get_snapshot_context(snapshot_id,
origin_url, timestamp, visit_id)
swh_type = snapshot_context['swh_type']
origin_info = snapshot_context['origin_info']
branches = snapshot_context['branches']
releases = snapshot_context['releases']
url_args = snapshot_context['url_args']
query_params = snapshot_context['query_params']
if snapshot_context['visit_info']:
timestamp = format_utc_iso_date(snapshot_context['visit_info']['date'],
'%Y-%m-%dT%H:%M:%SZ')
snapshot_context['timestamp'] = format_utc_iso_date(
snapshot_context['visit_info']['date'])
browse_view_name = 'browse-' + swh_type + '-' + browse_context
root_sha1_git = None
revision_id = request.GET.get('revision', None)
release_name = request.GET.get('release', None)
release_id = None
branch_name = None
- snapshot_size = snapshot_context['snapshot_size']
- snapshot_total_size = sum(snapshot_size.values())
+ snapshot_sizes = snapshot_context['snapshot_sizes']
+ snapshot_total_size = sum(snapshot_sizes.values())
if snapshot_total_size and revision_id:
revision = service.lookup_revision(revision_id)
root_sha1_git = revision['directory']
branches.append({'name': revision_id,
'revision': revision_id,
'directory': root_sha1_git,
'url': None})
branch_name = revision_id
query_params['revision'] = revision_id
elif snapshot_total_size and release_name:
release = _get_release(releases, release_name,
snapshot_context['snapshot_id'])
try:
root_sha1_git = release['directory']
revision_id = release['target']
release_id = release['id']
query_params['release'] = release_name
except Exception:
_branch_not_found('release', release_name, snapshot_id,
- snapshot_size, origin_info, timestamp, visit_id)
+ snapshot_sizes, origin_info, timestamp, visit_id)
elif snapshot_total_size:
branch_name = request.GET.get('branch', None)
if branch_name:
query_params['branch'] = branch_name
branch = _get_branch(branches, branch_name or 'HEAD',
snapshot_context['snapshot_id'])
try:
branch_name = branch['name']
revision_id = branch['revision']
root_sha1_git = branch['directory']
except Exception:
_branch_not_found('branch', branch_name, snapshot_id,
- snapshot_size, origin_info, timestamp, visit_id)
+ snapshot_sizes, origin_info, timestamp, visit_id)
for b in branches:
branch_url_args = dict(url_args)
branch_query_params = dict(query_params)
if 'release' in branch_query_params:
del branch_query_params['release']
branch_query_params['branch'] = b['name']
if path:
b['path'] = path
branch_url_args['path'] = path
b['url'] = reverse(browse_view_name,
url_args=branch_url_args,
query_params=branch_query_params)
for r in releases:
release_url_args = dict(url_args)
release_query_params = dict(query_params)
if 'branch' in release_query_params:
del release_query_params['branch']
release_query_params['release'] = r['name']
if path:
r['path'] = path
release_url_args['path'] = path
r['url'] = reverse(browse_view_name,
url_args=release_url_args,
query_params=release_query_params)
snapshot_context['query_params'] = query_params
snapshot_context['root_sha1_git'] = root_sha1_git
snapshot_context['revision_id'] = revision_id
snapshot_context['branch'] = branch_name
snapshot_context['release'] = release_name
snapshot_context['release_id'] = release_id
return snapshot_context
def browse_snapshot_directory(request, snapshot_id=None,
origin_url=None, timestamp=None, path=None):
"""
Django view implementation for browsing a directory in a snapshot context.
"""
try:
snapshot_context = _process_snapshot_request(
request, snapshot_id, origin_url,
timestamp, path, browse_context='directory')
root_sha1_git = snapshot_context['root_sha1_git']
sha1_git = root_sha1_git
if root_sha1_git and path:
dir_info = service.lookup_directory_with_path(root_sha1_git, path)
sha1_git = dir_info['target']
dirs = []
files = []
if sha1_git:
dirs, files = get_directory_entries(sha1_git)
except Exception as exc:
return handle_view_exception(request, exc)
swh_type = snapshot_context['swh_type']
origin_info = snapshot_context['origin_info']
visit_info = snapshot_context['visit_info']
url_args = snapshot_context['url_args']
query_params = snapshot_context['query_params']
revision_id = snapshot_context['revision_id']
snapshot_id = snapshot_context['snapshot_id']
path_info = gen_path_info(path)
browse_view_name = 'browse-' + swh_type + '-directory'
breadcrumbs = []
if root_sha1_git:
breadcrumbs.append({'name': root_sha1_git[:7],
'url': reverse(browse_view_name,
url_args=url_args,
query_params=query_params)})
for pi in path_info:
bc_url_args = dict(url_args)
bc_url_args['path'] = pi['path']
breadcrumbs.append({'name': pi['name'],
'url': reverse(browse_view_name,
url_args=bc_url_args,
query_params=query_params)})
path = '' if path is None else (path + '/')
for d in dirs:
if d['type'] == 'rev':
d['url'] = reverse('browse-revision',
url_args={'sha1_git': d['target']})
else:
bc_url_args = dict(url_args)
bc_url_args['path'] = path + d['name']
d['url'] = reverse(browse_view_name,
url_args=bc_url_args,
query_params=query_params)
sum_file_sizes = 0
readmes = {}
browse_view_name = 'browse-' + swh_type + '-content'
for f in files:
bc_url_args = dict(url_args)
bc_url_args['path'] = path + f['name']
f['url'] = reverse(browse_view_name,
url_args=bc_url_args,
query_params=query_params)
if f['length'] is not None:
sum_file_sizes += f['length']
f['length'] = filesizeformat(f['length'])
if f['name'].lower().startswith('readme'):
readmes[f['name']] = f['checksums']['sha1']
readme_name, readme_url, readme_html = get_readme_to_display(readmes)
browse_view_name = 'browse-' + swh_type + '-log'
history_url = None
if snapshot_id != _empty_snapshot_id:
history_url = reverse(browse_view_name,
url_args=url_args,
query_params=query_params)
nb_files = None
nb_dirs = None
dir_path = None
if root_sha1_git:
nb_files = len(files)
nb_dirs = len(dirs)
sum_file_sizes = filesizeformat(sum_file_sizes)
dir_path = '/' + path
browse_dir_link = gen_directory_link(sha1_git)
browse_rev_link = gen_revision_link(revision_id)
browse_snp_link = gen_snapshot_link(snapshot_id)
dir_metadata = {"directory": sha1_git,
"context-independent directory": browse_dir_link,
"number of regular files": nb_files,
"number of subdirectories": nb_dirs,
"sum of regular file sizes": sum_file_sizes,
"path": dir_path,
"revision": revision_id,
"context-independent revision": browse_rev_link,
"snapshot": snapshot_id,
"context-independent snapshot": browse_snp_link}
if origin_info:
dir_metadata['origin url'] = origin_info['url']
dir_metadata['origin visit date'] = format_utc_iso_date(
visit_info['date'])
dir_metadata['origin visit type'] = visit_info['type']
vault_cooking = {
'directory_context': True,
'directory_id': sha1_git,
'revision_context': True,
'revision_id': revision_id
}
swh_objects = [{'type': 'directory',
'id': sha1_git},
{'type': 'revision',
'id': revision_id},
{'type': 'snapshot',
'id': snapshot_id}]
release_id = snapshot_context['release_id']
if release_id:
swh_objects.append({'type': 'release',
'id': release_id})
browse_rel_link = gen_release_link(release_id)
dir_metadata['release'] = release_id
dir_metadata['context-independent release'] = browse_rel_link
swh_ids = get_swh_persistent_ids(swh_objects, snapshot_context)
dir_path = '/'.join([bc['name'] for bc in breadcrumbs]) + '/'
context_found = 'snapshot: %s' % snapshot_context['snapshot_id']
if origin_info:
context_found = 'origin: %s' % origin_info['url']
heading = ('Directory - %s - %s - %s' %
(dir_path, snapshot_context['branch'], context_found))
return render(request, 'browse/directory.html',
{'heading': heading,
'swh_object_name': 'Directory',
'swh_object_metadata': dir_metadata,
'dirs': dirs,
'files': files,
'breadcrumbs': breadcrumbs if root_sha1_git else [],
'top_right_link': {
'url': history_url,
'icon': swh_object_icons['revisions history'],
'text': 'History'
},
'readme_name': readme_name,
'readme_url': readme_url,
'readme_html': readme_html,
'snapshot_context': snapshot_context,
'vault_cooking': vault_cooking,
'show_actions_menu': True,
'swh_ids': swh_ids})
def browse_snapshot_content(request, snapshot_id=None,
origin_url=None, timestamp=None, path=None,
selected_language=None):
"""
Django view implementation for browsing a content in a snapshot context.
"""
try:
snapshot_context = _process_snapshot_request(request, snapshot_id,
origin_url,
timestamp, path,
browse_context='content')
root_sha1_git = snapshot_context['root_sha1_git']
sha1_git = None
query_string = None
content_data = None
directory_id = None
split_path = path.split('/')
filename = split_path[-1]
filepath = path[:-len(filename)]
if root_sha1_git:
content_info = service.lookup_directory_with_path(root_sha1_git,
path)
sha1_git = content_info['target']
query_string = 'sha1_git:' + sha1_git
content_data = request_content(query_string,
raise_if_unavailable=False)
if filepath:
dir_info = service.lookup_directory_with_path(root_sha1_git,
filepath)
directory_id = dir_info['target']
else:
directory_id = root_sha1_git
except Exception as exc:
return handle_view_exception(request, exc)
swh_type = snapshot_context['swh_type']
url_args = snapshot_context['url_args']
query_params = snapshot_context['query_params']
revision_id = snapshot_context['revision_id']
origin_info = snapshot_context['origin_info']
visit_info = snapshot_context['visit_info']
snapshot_id = snapshot_context['snapshot_id']
content = None
language = None
mimetype = None
if content_data and content_data['raw_data'] is not None:
content_display_data = prepare_content_for_display(
content_data['raw_data'], content_data['mimetype'], path)
content = content_display_data['content_data']
language = content_display_data['language']
mimetype = content_display_data['mimetype']
# Override language with user-selected language
if selected_language is not None:
language = selected_language
available_languages = None
if mimetype and 'text/' in mimetype:
available_languages = highlightjs.get_supported_languages()
browse_view_name = 'browse-' + swh_type + '-directory'
breadcrumbs = []
path_info = gen_path_info(filepath)
if root_sha1_git:
breadcrumbs.append({'name': root_sha1_git[:7],
'url': reverse(browse_view_name,
url_args=url_args,
query_params=query_params)})
for pi in path_info:
bc_url_args = dict(url_args)
bc_url_args['path'] = pi['path']
breadcrumbs.append({'name': pi['name'],
'url': reverse(browse_view_name,
url_args=bc_url_args,
query_params=query_params)})
breadcrumbs.append({'name': filename,
'url': None})
browse_content_link = gen_content_link(sha1_git)
content_raw_url = None
if query_string:
content_raw_url = reverse('browse-content-raw',
url_args={'query_string': query_string},
query_params={'filename': filename})
browse_rev_link = gen_revision_link(revision_id)
browse_dir_link = gen_directory_link(directory_id)
content_metadata = {
'context-independent content': browse_content_link,
'path': None,
'filename': None,
'directory': directory_id,
'context-independent directory': browse_dir_link,
'revision': revision_id,
'context-independent revision': browse_rev_link,
'snapshot': snapshot_id
}
cnt_sha1_git = None
content_size = None
error_code = 200
error_description = ''
error_message = ''
if content_data:
for checksum in content_data['checksums'].keys():
content_metadata[checksum] = content_data['checksums'][checksum]
content_metadata['mimetype'] = content_data['mimetype']
content_metadata['encoding'] = content_data['encoding']
content_metadata['size'] = filesizeformat(content_data['length'])
content_metadata['language'] = content_data['language']
content_metadata['licenses'] = content_data['licenses']
content_metadata['path'] = '/' + filepath
content_metadata['filename'] = filename
cnt_sha1_git = content_data['checksums']['sha1_git']
content_size = content_data['length']
error_code = content_data['error_code']
error_message = content_data['error_message']
error_description = content_data['error_description']
if origin_info:
content_metadata['origin url'] = origin_info['url']
content_metadata['origin visit date'] = format_utc_iso_date(
visit_info['date'])
content_metadata['origin visit type'] = visit_info['type']
browse_snapshot_link = gen_snapshot_link(snapshot_id)
content_metadata['context-independent snapshot'] = browse_snapshot_link
swh_objects = [{'type': 'content',
'id': cnt_sha1_git},
{'type': 'directory',
'id': directory_id},
{'type': 'revision',
'id': revision_id},
{'type': 'snapshot',
'id': snapshot_id}]
release_id = snapshot_context['release_id']
if release_id:
swh_objects.append({'type': 'release',
'id': release_id})
browse_rel_link = gen_release_link(release_id)
content_metadata['release'] = release_id
content_metadata['context-independent release'] = browse_rel_link
swh_ids = get_swh_persistent_ids(swh_objects, snapshot_context)
content_path = '/'.join([bc['name'] for bc in breadcrumbs])
context_found = 'snapshot: %s' % snapshot_context['snapshot_id']
if origin_info:
context_found = 'origin: %s' % origin_info['url']
heading = ('Content - %s - %s - %s' %
(content_path, snapshot_context['branch'], context_found))
return render(request, 'browse/content.html',
{'heading': heading,
'swh_object_name': 'Content',
'swh_object_metadata': content_metadata,
'content': content,
'content_size': content_size,
'max_content_size': content_display_max_size,
'mimetype': mimetype,
'language': language,
'available_languages': available_languages,
'breadcrumbs': breadcrumbs if root_sha1_git else [],
'top_right_link': {
'url': content_raw_url,
'icon': swh_object_icons['content'],
'text': 'Raw File'
},
'snapshot_context': snapshot_context,
'vault_cooking': None,
'show_actions_menu': True,
'swh_ids': swh_ids,
'error_code': error_code,
'error_message': error_message,
'error_description': error_description},
status=error_code)
PER_PAGE = 100
def browse_snapshot_log(request, snapshot_id=None,
origin_url=None, timestamp=None):
"""
Django view implementation for browsing a revision history in a
snapshot context.
"""
try:
snapshot_context = _process_snapshot_request(
request, snapshot_id, origin_url, timestamp, browse_context='log')
revision_id = snapshot_context['revision_id']
per_page = int(request.GET.get('per_page', PER_PAGE))
offset = int(request.GET.get('offset', 0))
revs_ordering = request.GET.get('revs_ordering', 'committer_date')
session_key = 'rev_%s_log_ordering_%s' % (revision_id, revs_ordering)
rev_log_session = request.session.get(session_key, None)
rev_log = []
revs_walker_state = None
if rev_log_session:
rev_log = rev_log_session['rev_log']
revs_walker_state = rev_log_session['revs_walker_state']
if len(rev_log) < offset+per_page:
revs_walker = service.get_revisions_walker(
revs_ordering, revision_id,
max_revs=offset+per_page+1,
state=revs_walker_state)
rev_log += [rev['id'] for rev in revs_walker]
revs_walker_state = revs_walker.export_state()
revs = rev_log[offset:offset+per_page]
revision_log = service.lookup_revision_multiple(revs)
request.session[session_key] = {
'rev_log': rev_log,
'revs_walker_state': revs_walker_state
}
except Exception as exc:
return handle_view_exception(request, exc)
swh_type = snapshot_context['swh_type']
origin_info = snapshot_context['origin_info']
visit_info = snapshot_context['visit_info']
url_args = snapshot_context['url_args']
query_params = snapshot_context['query_params']
snapshot_id = snapshot_context['snapshot_id']
query_params['per_page'] = per_page
revs_ordering = request.GET.get('revs_ordering', '')
query_params['revs_ordering'] = revs_ordering
browse_view_name = 'browse-' + swh_type + '-log'
prev_log_url = None
if len(rev_log) > offset + per_page:
query_params['offset'] = offset + per_page
prev_log_url = reverse(browse_view_name,
url_args=url_args,
query_params=query_params)
next_log_url = None
if offset != 0:
query_params['offset'] = offset - per_page
next_log_url = reverse(browse_view_name,
url_args=url_args,
query_params=query_params)
revision_log_data = format_log_entries(revision_log, per_page,
snapshot_context)
browse_rev_link = gen_revision_link(revision_id)
browse_log_link = gen_revision_log_link(revision_id)
browse_snp_link = gen_snapshot_link(snapshot_id)
revision_metadata = {
'context-independent revision': browse_rev_link,
'context-independent revision history': browse_log_link,
'context-independent snapshot': browse_snp_link,
'snapshot': snapshot_id
}
if origin_info:
revision_metadata['origin url'] = origin_info['url']
revision_metadata['origin visit date'] = format_utc_iso_date(
visit_info['date'])
revision_metadata['origin visit type'] = visit_info['type']
swh_objects = [{'type': 'revision',
'id': revision_id},
{'type': 'snapshot',
'id': snapshot_id}]
release_id = snapshot_context['release_id']
if release_id:
swh_objects.append({'type': 'release',
'id': release_id})
browse_rel_link = gen_release_link(release_id)
revision_metadata['release'] = release_id
revision_metadata['context-independent release'] = browse_rel_link
swh_ids = get_swh_persistent_ids(swh_objects, snapshot_context)
context_found = 'snapshot: %s' % snapshot_context['snapshot_id']
if origin_info:
context_found = 'origin: %s' % origin_info['url']
heading = ('Revision history - %s - %s' %
(snapshot_context['branch'], context_found))
return render(request, 'browse/revision-log.html',
{'heading': heading,
'swh_object_name': 'Revisions history',
'swh_object_metadata': revision_metadata,
'revision_log': revision_log_data,
'revs_ordering': revs_ordering,
'next_log_url': next_log_url,
'prev_log_url': prev_log_url,
'breadcrumbs': None,
'top_right_link': None,
'snapshot_context': snapshot_context,
'vault_cooking': None,
'show_actions_menu': True,
'swh_ids': swh_ids})
def browse_snapshot_branches(request, snapshot_id=None,
origin_url=None, timestamp=None):
"""
Django view implementation for browsing a list of branches in a snapshot
context.
"""
try:
snapshot_context = _process_snapshot_request(request, snapshot_id,
origin_url, timestamp)
branches_bc = request.GET.get('branches_breadcrumbs', '')
branches_bc = branches_bc.split(',') if branches_bc else []
branches_from = branches_bc[-1] if branches_bc else ''
swh_type = snapshot_context['swh_type']
origin_info = snapshot_context['origin_info']
url_args = snapshot_context['url_args']
query_params = snapshot_context['query_params']
browse_view_name = 'browse-' + swh_type + '-directory'
snapshot = service.lookup_snapshot(snapshot_context['snapshot_id'],
branches_from, PER_PAGE+1,
target_types=['revision', 'alias'])
displayed_branches, _ = process_snapshot_branches(snapshot)
except Exception as exc:
return handle_view_exception(request, exc)
for branch in displayed_branches:
if snapshot_id:
revision_url = reverse('browse-revision',
url_args={'sha1_git': branch['revision']},
query_params={'snapshot_id': snapshot_id})
else:
revision_url = reverse('browse-revision',
url_args={'sha1_git': branch['revision']},
query_params={'origin': origin_info['url']})
query_params['branch'] = branch['name']
directory_url = reverse(browse_view_name,
url_args=url_args,
query_params=query_params)
del query_params['branch']
branch['revision_url'] = revision_url
branch['directory_url'] = directory_url
browse_view_name = 'browse-' + swh_type + '-branches'
prev_branches_url = None
next_branches_url = None
if branches_bc:
query_params_prev = dict(query_params)
query_params_prev['branches_breadcrumbs'] = ','.join(branches_bc[:-1])
prev_branches_url = reverse(browse_view_name, url_args=url_args,
query_params=query_params_prev)
elif branches_from:
prev_branches_url = reverse(browse_view_name, url_args=url_args,
query_params=query_params)
if len(displayed_branches) > PER_PAGE:
query_params_next = dict(query_params)
next_branch = displayed_branches[-1]['name']
del displayed_branches[-1]
branches_bc.append(next_branch)
query_params_next['branches_breadcrumbs'] = ','.join(branches_bc)
next_branches_url = reverse(browse_view_name, url_args=url_args,
query_params=query_params_next)
heading = 'Branches - '
if origin_info:
heading += 'origin: %s' % origin_info['url']
else:
heading += 'snapshot: %s' % snapshot_id
return render(request, 'browse/branches.html',
{'heading': heading,
'swh_object_name': 'Branches',
'swh_object_metadata': {},
'top_right_link': None,
'displayed_branches': displayed_branches,
'prev_branches_url': prev_branches_url,
'next_branches_url': next_branches_url,
'snapshot_context': snapshot_context})
def browse_snapshot_releases(request, snapshot_id=None,
origin_url=None, timestamp=None):
"""
Django view implementation for browsing a list of releases in a snapshot
context.
"""
try:
snapshot_context = _process_snapshot_request(request, snapshot_id,
origin_url, timestamp)
rel_bc = request.GET.get('releases_breadcrumbs', '')
rel_bc = rel_bc.split(',') if rel_bc else []
rel_from = rel_bc[-1] if rel_bc else ''
swh_type = snapshot_context['swh_type']
origin_info = snapshot_context['origin_info']
url_args = snapshot_context['url_args']
query_params = snapshot_context['query_params']
snapshot = service.lookup_snapshot(snapshot_context['snapshot_id'],
rel_from, PER_PAGE+1,
target_types=['release', 'alias'])
_, displayed_releases = process_snapshot_branches(snapshot)
except Exception as exc:
return handle_view_exception(request, exc)
for release in displayed_releases:
if snapshot_id:
query_params_tgt = {'snapshot_id': snapshot_id}
else:
query_params_tgt = {'origin': origin_info['url']}
release_url = reverse('browse-release',
url_args={'sha1_git': release['id']},
query_params=query_params_tgt)
target_url = ''
if release['target_type'] == 'revision':
target_url = reverse('browse-revision',
url_args={'sha1_git': release['target']},
query_params=query_params_tgt)
elif release['target_type'] == 'directory':
target_url = reverse('browse-directory',
url_args={'sha1_git': release['target']},
query_params=query_params_tgt)
elif release['target_type'] == 'content':
target_url = reverse('browse-content',
url_args={'query_string': release['target']},
query_params=query_params_tgt)
elif release['target_type'] == 'release':
target_url = reverse('browse-release',
url_args={'sha1_git': release['target']},
query_params=query_params_tgt)
release['release_url'] = release_url
release['target_url'] = target_url
browse_view_name = 'browse-' + swh_type + '-releases'
prev_releases_url = None
next_releases_url = None
if rel_bc:
query_params_prev = dict(query_params)
query_params_prev['releases_breadcrumbs'] = ','.join(rel_bc[:-1])
prev_releases_url = reverse(browse_view_name, url_args=url_args,
query_params=query_params_prev)
elif rel_from:
prev_releases_url = reverse(browse_view_name, url_args=url_args,
query_params=query_params)
if len(displayed_releases) > PER_PAGE:
query_params_next = dict(query_params)
next_rel = displayed_releases[-1]['branch_name']
del displayed_releases[-1]
rel_bc.append(next_rel)
query_params_next['releases_breadcrumbs'] = ','.join(rel_bc)
next_releases_url = reverse(browse_view_name, url_args=url_args,
query_params=query_params_next)
heading = 'Releases - '
if origin_info:
heading += 'origin: %s' % origin_info['url']
else:
heading += 'snapshot: %s' % snapshot_id
return render(request, 'browse/releases.html',
{'heading': heading,
'top_panel_visible': False,
'top_panel_collapsible': False,
'swh_object_name': 'Releases',
'swh_object_metadata': {},
'top_right_link': None,
'displayed_releases': displayed_releases,
'prev_releases_url': prev_releases_url,
'next_releases_url': next_releases_url,
'snapshot_context': snapshot_context,
'vault_cooking': None,
'show_actions_menu': False})
diff --git a/swh/web/common/service.py b/swh/web/common/service.py
index e2de756f..fbb0889e 100644
--- a/swh/web/common/service.py
+++ b/swh/web/common/service.py
@@ -1,1147 +1,1147 @@
# Copyright (C) 2015-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import os
from collections import defaultdict
from typing import Any, Dict
from swh.model import hashutil
from swh.storage.algos import diff, revisions_walker
from swh.model.identifiers import (
CONTENT, DIRECTORY, RELEASE, REVISION, SNAPSHOT
)
from swh.web.common import converters
from swh.web.common import query
from swh.web.common.exc import BadInputExc, NotFoundExc
from swh.web.common.origin_visits import get_origin_visit
from swh.web import config
storage = config.storage()
vault = config.vault()
idx_storage = config.indexer_storage()
MAX_LIMIT = 50 # Top limit the users can ask for
def _first_element(l):
"""Returns the first element in the provided list or None
if it is empty or None"""
return next(iter(l or []), None)
def lookup_multiple_hashes(hashes):
"""Lookup the passed hashes in a single DB connection, using batch
processing.
Args:
An array of {filename: X, sha1: Y}, string X, hex sha1 string Y.
Returns:
The same array with elements updated with elem['found'] = true if
the hash is present in storage, elem['found'] = false if not.
"""
hashlist = [hashutil.hash_to_bytes(elem['sha1']) for elem in hashes]
content_missing = storage.content_missing_per_sha1(hashlist)
missing = [hashutil.hash_to_hex(x) for x in content_missing]
for x in hashes:
x.update({'found': True})
for h in hashes:
if h['sha1'] in missing:
h['found'] = False
return hashes
def lookup_expression(expression, last_sha1, per_page):
"""Lookup expression in raw content.
Args:
expression (str): An expression to lookup through raw indexed
content
last_sha1 (str): Last sha1 seen
per_page (int): Number of results per page
Yields:
ctags whose content match the expression
"""
limit = min(per_page, MAX_LIMIT)
ctags = idx_storage.content_ctags_search(expression,
last_sha1=last_sha1,
limit=limit)
for ctag in ctags:
ctag = converters.from_swh(ctag, hashess={'id'})
ctag['sha1'] = ctag['id']
ctag.pop('id')
yield ctag
def lookup_hash(q):
"""Checks if the storage contains a given content checksum
Args: query string of the form
Returns: Dict with key found containing the hash info if the
hash is present, None if not.
"""
algo, hash = query.parse_hash(q)
found = _first_element(storage.content_find({algo: hash}))
return {'found': converters.from_content(found),
'algo': algo}
def search_hash(q):
"""Checks if the storage contains a given content checksum
Args: query string of the form
Returns: Dict with key found to True or False, according to
whether the checksum is present or not
"""
algo, hash = query.parse_hash(q)
found = _first_element(storage.content_find({algo: hash}))
return {'found': found is not None}
def _lookup_content_sha1(q):
"""Given a possible input, query for the content's sha1.
Args:
q: query string of the form
Returns:
binary sha1 if found or None
"""
algo, hash = query.parse_hash(q)
if algo != 'sha1':
hashes = _first_element(storage.content_find({algo: hash}))
if not hashes:
return None
return hashes['sha1']
return hash
def lookup_content_ctags(q):
"""Return ctags information from a specified content.
Args:
q: query string of the form
Yields:
ctags information (dict) list if the content is found.
"""
sha1 = _lookup_content_sha1(q)
if not sha1:
return None
ctags = list(idx_storage.content_ctags_get([sha1]))
if not ctags:
return None
for ctag in ctags:
yield converters.from_swh(ctag, hashess={'id'})
def lookup_content_filetype(q):
"""Return filetype information from a specified content.
Args:
q: query string of the form
Yields:
filetype information (dict) list if the content is found.
"""
sha1 = _lookup_content_sha1(q)
if not sha1:
return None
filetype = _first_element(list(idx_storage.content_mimetype_get([sha1])))
if not filetype:
return None
return converters.from_filetype(filetype)
def lookup_content_language(q):
"""Return language information from a specified content.
Args:
q: query string of the form
Yields:
language information (dict) list if the content is found.
"""
sha1 = _lookup_content_sha1(q)
if not sha1:
return None
lang = _first_element(list(idx_storage.content_language_get([sha1])))
if not lang:
return None
return converters.from_swh(lang, hashess={'id'})
def lookup_content_license(q):
"""Return license information from a specified content.
Args:
q: query string of the form
Yields:
license information (dict) list if the content is found.
"""
sha1 = _lookup_content_sha1(q)
if not sha1:
return None
lic = _first_element(idx_storage.content_fossology_license_get([sha1]))
if not lic:
return None
return converters.from_swh({'id': sha1, 'facts': lic[sha1]},
hashess={'id'})
def lookup_origin(origin):
"""Return information about the origin matching dict origin.
Args:
origin: origin's dict with 'url' key
Returns:
origin information as dict.
"""
origin_info = storage.origin_get(origin)
if not origin_info:
msg = 'Origin with url %s not found!' % origin['url']
raise NotFoundExc(msg)
return converters.from_origin(origin_info)
def lookup_origins(origin_from=1, origin_count=100):
"""Get list of archived software origins in a paginated way.
Origins are sorted by id before returning them
Args:
origin_from (int): The minimum id of the origins to return
origin_count (int): The maximum number of origins to return
Yields:
origins information as dicts
"""
origins = storage.origin_get_range(origin_from, origin_count)
return map(converters.from_origin, origins)
def search_origin(url_pattern, offset=0, limit=50, regexp=False,
with_visit=False):
"""Search for origins whose urls contain a provided string pattern
or match a provided regular expression.
Args:
url_pattern: the string pattern to search for in origin urls
offset: number of found origins to skip before returning results
limit: the maximum number of found origins to return
Returns:
list of origin information as dict.
"""
origins = storage.origin_search(url_pattern, offset, limit, regexp,
with_visit)
return map(converters.from_origin, origins)
def search_origin_metadata(fulltext, limit=50):
"""Search for origins whose metadata match a provided string pattern.
Args:
fulltext: the string pattern to search for in origin metadata
offset: number of found origins to skip before returning results
limit: the maximum number of found origins to return
Returns:
list of origin metadata as dict.
"""
matches = idx_storage.origin_intrinsic_metadata_search_fulltext(
conjunction=[fulltext], limit=limit)
results = []
for match in matches:
match['from_revision'] = hashutil.hash_to_hex(match['from_revision'])
origin = storage.origin_get({'url': match['id']})
del match['id']
result = converters.from_origin(origin)
if result:
result['metadata'] = match
results.append(result)
return results
def lookup_origin_intrinsic_metadata(origin_dict):
"""Return intrinsic metadata for origin whose origin matches given
origin.
Args:
origin_dict: origin's dict with keys ('type' AND 'url')
Returns:
origin metadata.
"""
origin_info = storage.origin_get(origin_dict)
if not origin_info:
msg = 'Origin with url %s not found!' % origin_dict['url']
raise NotFoundExc(msg)
origins = [origin_info['url']]
match = _first_element(
idx_storage.origin_intrinsic_metadata_get(origins))
result = {}
if match:
result = match['metadata']
return result
def _to_sha1_bin(sha1_hex):
_, sha1_git_bin = query.parse_hash_with_algorithms_or_throws(
sha1_hex,
['sha1'], # HACK: sha1_git really
'Only sha1_git is supported.')
return sha1_git_bin
def _check_directory_exists(sha1_git, sha1_git_bin):
if len(list(storage.directory_missing([sha1_git_bin]))):
raise NotFoundExc('Directory with sha1_git %s not found' % sha1_git)
def lookup_directory(sha1_git):
"""Return information about the directory with id sha1_git.
Args:
sha1_git as string
Returns:
directory information as dict.
"""
empty_dir_sha1 = '4b825dc642cb6eb9a060e54bf8d69288fbee4904'
if sha1_git == empty_dir_sha1:
return []
sha1_git_bin = _to_sha1_bin(sha1_git)
_check_directory_exists(sha1_git, sha1_git_bin)
directory_entries = storage.directory_ls(sha1_git_bin)
return map(converters.from_directory_entry, directory_entries)
def lookup_directory_with_path(sha1_git, path_string):
"""Return directory information for entry with path path_string w.r.t.
root directory pointed by directory_sha1_git
Args:
- directory_sha1_git: sha1_git corresponding to the directory
to which we append paths to (hopefully) find the entry
- the relative path to the entry starting from the directory pointed by
directory_sha1_git
Raises:
NotFoundExc if the directory entry is not found
"""
sha1_git_bin = _to_sha1_bin(sha1_git)
_check_directory_exists(sha1_git, sha1_git_bin)
paths = path_string.strip(os.path.sep).split(os.path.sep)
queried_dir = storage.directory_entry_get_by_path(
sha1_git_bin, list(map(lambda p: p.encode('utf-8'), paths)))
if not queried_dir:
raise NotFoundExc(('Directory entry with path %s from %s not found') %
(path_string, sha1_git))
return converters.from_directory_entry(queried_dir)
def lookup_release(release_sha1_git):
"""Return information about the release with sha1 release_sha1_git.
Args:
release_sha1_git: The release's sha1 as hexadecimal
Returns:
Release information as dict.
Raises:
ValueError if the identifier provided is not of sha1 nature.
"""
sha1_git_bin = _to_sha1_bin(release_sha1_git)
release = _first_element(storage.release_get([sha1_git_bin]))
if not release:
raise NotFoundExc('Release with sha1_git %s not found.'
% release_sha1_git)
return converters.from_release(release)
def lookup_release_multiple(sha1_git_list):
"""Return information about the revisions identified with
their sha1_git identifiers.
Args:
sha1_git_list: A list of revision sha1_git identifiers
Returns:
Release information as dict.
Raises:
ValueError if the identifier provided is not of sha1 nature.
"""
sha1_bin_list = (_to_sha1_bin(sha1_git) for sha1_git in sha1_git_list)
releases = storage.release_get(sha1_bin_list) or []
return (converters.from_release(r) for r in releases)
def lookup_revision(rev_sha1_git):
"""Return information about the revision with sha1 revision_sha1_git.
Args:
revision_sha1_git: The revision's sha1 as hexadecimal
Returns:
Revision information as dict.
Raises:
ValueError if the identifier provided is not of sha1 nature.
NotFoundExc if there is no revision with the provided sha1_git.
"""
sha1_git_bin = _to_sha1_bin(rev_sha1_git)
revision = _first_element(storage.revision_get([sha1_git_bin]))
if not revision:
raise NotFoundExc('Revision with sha1_git %s not found.'
% rev_sha1_git)
return converters.from_revision(revision)
def lookup_revision_multiple(sha1_git_list):
"""Return information about the revisions identified with
their sha1_git identifiers.
Args:
sha1_git_list: A list of revision sha1_git identifiers
Returns:
Generator of revisions information as dict.
Raises:
ValueError if the identifier provided is not of sha1 nature.
"""
sha1_bin_list = (_to_sha1_bin(sha1_git) for sha1_git in sha1_git_list)
revisions = storage.revision_get(sha1_bin_list) or []
return (converters.from_revision(r) for r in revisions)
def lookup_revision_message(rev_sha1_git):
"""Return the raw message of the revision with sha1 revision_sha1_git.
Args:
revision_sha1_git: The revision's sha1 as hexadecimal
Returns:
Decoded revision message as dict {'message': }
Raises:
ValueError if the identifier provided is not of sha1 nature.
NotFoundExc if the revision is not found, or if it has no message
"""
sha1_git_bin = _to_sha1_bin(rev_sha1_git)
revision = _first_element(storage.revision_get([sha1_git_bin]))
if not revision:
raise NotFoundExc('Revision with sha1_git %s not found.'
% rev_sha1_git)
if 'message' not in revision:
raise NotFoundExc('No message for revision with sha1_git %s.'
% rev_sha1_git)
res = {'message': revision['message']}
return res
def _lookup_revision_id_by(origin, branch_name, timestamp):
def _get_snapshot_branch(snapshot, branch_name):
snapshot = lookup_snapshot(visit['snapshot'],
branches_from=branch_name,
branches_count=10)
branch = None
if branch_name in snapshot['branches']:
branch = snapshot['branches'][branch_name]
return branch
if isinstance(origin, int):
origin = {'id': origin}
elif isinstance(origin, str):
origin = {'url': origin}
else:
raise TypeError('"origin" must be an int or a string.')
visit = get_origin_visit(origin, visit_ts=timestamp)
branch = _get_snapshot_branch(visit['snapshot'], branch_name)
rev_id = None
if branch and branch['target_type'] == 'revision':
rev_id = branch['target']
elif branch and branch['target_type'] == 'alias':
branch = _get_snapshot_branch(visit['snapshot'], branch['target'])
if branch and branch['target_type'] == 'revision':
rev_id = branch['target']
if not rev_id:
raise NotFoundExc('Revision for origin %s and branch %s not found.'
% (origin.get('url'), branch_name))
return rev_id
def lookup_revision_by(origin,
branch_name='HEAD',
timestamp=None):
"""Lookup revision by origin, snapshot branch name and visit timestamp.
If branch_name is not provided, lookup using 'HEAD' as default.
If timestamp is not provided, use the most recent.
Args:
origin (Union[int,str]): origin of the revision
branch_name (str): snapshot branch name
timestamp (str/int): origin visit time frame
Returns:
dict: The revision matching the criterions
Raises:
NotFoundExc if no revision corresponds to the criterion
"""
rev_id = _lookup_revision_id_by(origin, branch_name, timestamp)
return lookup_revision(rev_id)
def lookup_revision_log(rev_sha1_git, limit):
"""Lookup revision log by revision id.
Args:
rev_sha1_git (str): The revision's sha1 as hexadecimal
limit (int): the maximum number of revisions returned
Returns:
list: Revision log as list of revision dicts
Raises:
ValueError: if the identifier provided is not of sha1 nature.
NotFoundExc: if there is no revision with the provided sha1_git.
"""
lookup_revision(rev_sha1_git)
sha1_git_bin = _to_sha1_bin(rev_sha1_git)
revision_entries = storage.revision_log([sha1_git_bin], limit)
return map(converters.from_revision, revision_entries)
def lookup_revision_log_by(origin, branch_name, timestamp, limit):
"""Lookup revision by origin, snapshot branch name and visit timestamp.
Args:
origin (Union[int,str]): origin of the revision
branch_name (str): snapshot branch
timestamp (str/int): origin visit time frame
limit (int): the maximum number of revisions returned
Returns:
list: Revision log as list of revision dicts
Raises:
NotFoundExc: if no revision corresponds to the criterion
"""
rev_id = _lookup_revision_id_by(origin, branch_name, timestamp)
return lookup_revision_log(rev_id, limit)
def lookup_revision_with_context_by(origin, branch_name, timestamp,
sha1_git, limit=100):
"""Return information about revision sha1_git, limited to the
sub-graph of all transitive parents of sha1_git_root.
sha1_git_root being resolved through the lookup of a revision by origin,
branch_name and ts.
In other words, sha1_git is an ancestor of sha1_git_root.
Args:
- origin: origin of the revision.
- branch_name: revision's branch.
- timestamp: revision's time frame.
- sha1_git: one of sha1_git_root's ancestors.
- limit: limit the lookup to 100 revisions back.
Returns:
Pair of (root_revision, revision).
Information on sha1_git if it is an ancestor of sha1_git_root
including children leading to sha1_git_root
Raises:
- BadInputExc in case of unknown algo_hash or bad hash.
- NotFoundExc if either revision is not found or if sha1_git is not an
ancestor of sha1_git_root.
"""
rev_root_id = _lookup_revision_id_by(origin, branch_name, timestamp)
rev_root_id_bin = hashutil.hash_to_bytes(rev_root_id)
rev_root = _first_element(storage.revision_get([rev_root_id_bin]))
return (converters.from_revision(rev_root),
lookup_revision_with_context(rev_root, sha1_git, limit))
def lookup_revision_with_context(sha1_git_root, sha1_git, limit=100):
"""Return information about revision sha1_git, limited to the
sub-graph of all transitive parents of sha1_git_root.
In other words, sha1_git is an ancestor of sha1_git_root.
Args:
sha1_git_root: latest revision. The type is either a sha1 (as an hex
string) or a non converted dict.
sha1_git: one of sha1_git_root's ancestors
limit: limit the lookup to 100 revisions back
Returns:
Information on sha1_git if it is an ancestor of sha1_git_root
including children leading to sha1_git_root
Raises:
BadInputExc in case of unknown algo_hash or bad hash
NotFoundExc if either revision is not found or if sha1_git is not an
ancestor of sha1_git_root
"""
sha1_git_bin = _to_sha1_bin(sha1_git)
revision = _first_element(storage.revision_get([sha1_git_bin]))
if not revision:
raise NotFoundExc('Revision %s not found' % sha1_git)
if isinstance(sha1_git_root, str):
sha1_git_root_bin = _to_sha1_bin(sha1_git_root)
revision_root = _first_element(storage.revision_get([sha1_git_root_bin])) # noqa
if not revision_root:
raise NotFoundExc('Revision root %s not found' % sha1_git_root)
else:
sha1_git_root_bin = sha1_git_root['id']
revision_log = storage.revision_log([sha1_git_root_bin], limit)
parents = {}
children = defaultdict(list)
for rev in revision_log:
rev_id = rev['id']
parents[rev_id] = []
for parent_id in rev['parents']:
parents[rev_id].append(parent_id)
children[parent_id].append(rev_id)
if revision['id'] not in parents:
raise NotFoundExc('Revision %s is not an ancestor of %s' %
(sha1_git, sha1_git_root))
revision['children'] = children[revision['id']]
return converters.from_revision(revision)
def lookup_directory_with_revision(sha1_git, dir_path=None, with_data=False):
"""Return information on directory pointed by revision with sha1_git.
If dir_path is not provided, display top level directory.
Otherwise, display the directory pointed by dir_path (if it exists).
Args:
sha1_git: revision's hash.
dir_path: optional directory pointed to by that revision.
with_data: boolean that indicates to retrieve the raw data if the path
resolves to a content. Default to False (for the api)
Returns:
Information on the directory pointed to by that revision.
Raises:
BadInputExc in case of unknown algo_hash or bad hash.
NotFoundExc either if the revision is not found or the path referenced
does not exist.
NotImplementedError in case of dir_path exists but do not reference a
type 'dir' or 'file'.
"""
sha1_git_bin = _to_sha1_bin(sha1_git)
revision = _first_element(storage.revision_get([sha1_git_bin]))
if not revision:
raise NotFoundExc('Revision %s not found' % sha1_git)
dir_sha1_git_bin = revision['directory']
if dir_path:
paths = dir_path.strip(os.path.sep).split(os.path.sep)
entity = storage.directory_entry_get_by_path(
dir_sha1_git_bin, list(map(lambda p: p.encode('utf-8'), paths)))
if not entity:
raise NotFoundExc(
"Directory or File '%s' pointed to by revision %s not found"
% (dir_path, sha1_git))
else:
entity = {'type': 'dir', 'target': dir_sha1_git_bin}
if entity['type'] == 'dir':
directory_entries = storage.directory_ls(entity['target']) or []
return {'type': 'dir',
'path': '.' if not dir_path else dir_path,
'revision': sha1_git,
'content': list(map(converters.from_directory_entry,
directory_entries))}
elif entity['type'] == 'file': # content
content = _first_element(
storage.content_find({'sha1_git': entity['target']}))
if not content:
raise NotFoundExc('Content not found for revision %s'
% sha1_git)
if with_data:
c = _first_element(storage.content_get([content['sha1']]))
content['data'] = c['data']
return {'type': 'file',
'path': '.' if not dir_path else dir_path,
'revision': sha1_git,
'content': converters.from_content(content)}
elif entity['type'] == 'rev': # revision
revision = next(storage.revision_get([entity['target']]))
return {'type': 'rev',
'path': '.' if not dir_path else dir_path,
'revision': sha1_git,
'content': converters.from_revision(revision)}
else:
raise NotImplementedError('Entity of type %s not implemented.'
% entity['type'])
def lookup_content(q):
"""Lookup the content designed by q.
Args:
q: The release's sha1 as hexadecimal
Raises:
NotFoundExc if the requested content is not found
"""
algo, hash = query.parse_hash(q)
c = _first_element(storage.content_find({algo: hash}))
if not c:
raise NotFoundExc('Content with %s checksum equals to %s not found!' %
(algo, hashutil.hash_to_hex(hash)))
return converters.from_content(c)
def lookup_content_raw(q):
"""Lookup the content defined by q.
Args:
q: query string of the form
Returns:
dict with 'sha1' and 'data' keys.
data representing its raw data decoded.
Raises:
NotFoundExc if the requested content is not found or
if the content bytes are not available in the storage
"""
c = lookup_content(q)
content_sha1_bytes = hashutil.hash_to_bytes(c['checksums']['sha1'])
content = _first_element(storage.content_get([content_sha1_bytes]))
if not content:
algo, hash = query.parse_hash(q)
raise NotFoundExc('Bytes of content with %s checksum equals to %s '
'are not available!' %
(algo, hashutil.hash_to_hex(hash)))
return converters.from_content(content)
def stat_counters():
"""Return the stat counters for Software Heritage
Returns:
A dict mapping textual labels to integer values.
"""
return storage.stat_counters()
def _lookup_origin_visits(origin_url, last_visit=None, limit=10):
"""Yields the origin origins' visits.
Args:
origin_url (str): origin to list visits for
last_visit (int): last visit to lookup from
limit (int): Number of elements max to display
Yields:
Dictionaries of origin_visit for that origin
"""
limit = min(limit, MAX_LIMIT)
for visit in storage.origin_visit_get(
origin_url, last_visit=last_visit, limit=limit):
visit['origin'] = origin_url
yield visit
def lookup_origin_visits(origin, last_visit=None, per_page=10):
"""Yields the origin origins' visits.
Args:
origin: origin to list visits for
Yields:
Dictionaries of origin_visit for that origin
"""
visits = _lookup_origin_visits(origin, last_visit=last_visit,
limit=per_page)
for visit in visits:
yield converters.from_origin_visit(visit)
def lookup_origin_visit_latest(origin_url, require_snapshot):
"""Return the origin's latest visit
Args:
origin_url (str): origin to list visits for
require_snapshot (bool): filter out origins without a snapshot
Returns:
dict: The origin_visit concerned
"""
visit = storage.origin_visit_get_latest(
origin_url, require_snapshot=require_snapshot)
return converters.from_origin_visit(visit)
def lookup_origin_visit(origin_url, visit_id):
"""Return information about visit visit_id with origin origin.
Args:
origin (str): origin concerned by the visit
visit_id: the visit identifier to lookup
Yields:
The dict origin_visit concerned
"""
visit = storage.origin_visit_get_by(origin_url, visit_id)
if not visit:
raise NotFoundExc('Origin %s or its visit '
'with id %s not found!' % (origin_url, visit_id))
visit['origin'] = origin_url
return converters.from_origin_visit(visit)
-def lookup_snapshot_size(snapshot_id):
+def lookup_snapshot_sizes(snapshot_id):
"""Count the number of branches in the snapshot with the given id
Args:
snapshot_id (str): sha1 identifier of the snapshot
Returns:
dict: A dict whose keys are the target types of branches and
values their corresponding amount
"""
snapshot_id_bin = _to_sha1_bin(snapshot_id)
- snapshot_size = storage.snapshot_count_branches(snapshot_id_bin)
- if 'revision' not in snapshot_size:
- snapshot_size['revision'] = 0
- if 'release' not in snapshot_size:
- snapshot_size['release'] = 0
+ snapshot_sizes = storage.snapshot_count_branches(snapshot_id_bin)
+ if 'revision' not in snapshot_sizes:
+ snapshot_sizes['revision'] = 0
+ if 'release' not in snapshot_sizes:
+ snapshot_sizes['release'] = 0
# adjust revision / release count for display if aliases are defined
- if 'alias' in snapshot_size:
+ if 'alias' in snapshot_sizes:
aliases = lookup_snapshot(snapshot_id,
- branches_count=snapshot_size['alias'],
+ branches_count=snapshot_sizes['alias'],
target_types=['alias'])
for alias in aliases['branches'].values():
if lookup_snapshot(snapshot_id,
branches_from=alias['target'],
branches_count=1,
target_types=['revision']):
- snapshot_size['revision'] += 1
+ snapshot_sizes['revision'] += 1
else:
- snapshot_size['release'] += 1
- del snapshot_size['alias']
- return snapshot_size
+ snapshot_sizes['release'] += 1
+ del snapshot_sizes['alias']
+ return snapshot_sizes
def lookup_snapshot(snapshot_id, branches_from='', branches_count=1000,
target_types=None):
"""Return information about a snapshot, aka the list of named
branches found during a specific visit of an origin.
Args:
snapshot_id (str): sha1 identifier of the snapshot
branches_from (str): optional parameter used to skip branches
whose name is lesser than it before returning them
branches_count (int): optional parameter used to restrain
the amount of returned branches
target_types (list): optional parameter used to filter the
target types of branch to return (possible values that can be
contained in that list are `'content', 'directory',
'revision', 'release', 'snapshot', 'alias'`)
Returns:
A dict filled with the snapshot content.
"""
snapshot_id_bin = _to_sha1_bin(snapshot_id)
snapshot = storage.snapshot_get_branches(snapshot_id_bin,
branches_from.encode(),
branches_count, target_types)
if not snapshot:
raise NotFoundExc('Snapshot with id %s not found!' % snapshot_id)
return converters.from_snapshot(snapshot)
def lookup_latest_origin_snapshot(origin, allowed_statuses=None):
"""Return information about the latest snapshot of an origin.
.. warning:: At most 1000 branches contained in the snapshot
will be returned for performance reasons.
Args:
origin: URL or integer identifier of the origin
allowed_statuses: list of visit statuses considered
to find the latest snapshot for the visit. For instance,
``allowed_statuses=['full']`` will only consider visits that
have successfully run to completion.
Returns:
A dict filled with the snapshot content.
"""
snapshot = storage.snapshot_get_latest(origin, allowed_statuses)
return converters.from_snapshot(snapshot)
def lookup_revision_through(revision, limit=100):
"""Retrieve a revision from the criterion stored in revision dictionary.
Args:
revision: Dictionary of criterion to lookup the revision with.
Here are the supported combination of possible values:
- origin_url, branch_name, ts, sha1_git
- origin_url, branch_name, ts
- sha1_git_root, sha1_git
- sha1_git
Returns:
None if the revision is not found or the actual revision.
"""
if (
'origin_url' in revision and
'branch_name' in revision and
'ts' in revision and
'sha1_git' in revision):
return lookup_revision_with_context_by(revision['origin_url'],
revision['branch_name'],
revision['ts'],
revision['sha1_git'],
limit)
if (
'origin_url' in revision and
'branch_name' in revision and
'ts' in revision):
return lookup_revision_by(revision['origin_url'],
revision['branch_name'],
revision['ts'])
if (
'sha1_git_root' in revision and
'sha1_git' in revision):
return lookup_revision_with_context(revision['sha1_git_root'],
revision['sha1_git'],
limit)
if 'sha1_git' in revision:
return lookup_revision(revision['sha1_git'])
# this should not happen
raise NotImplementedError('Should not happen!')
def lookup_directory_through_revision(revision, path=None,
limit=100, with_data=False):
"""Retrieve the directory information from the revision.
Args:
revision: dictionary of criterion representing a revision to lookup
path: directory's path to lookup.
limit: optional query parameter to limit the revisions log (default to
100). For now, note that this limit could impede the transitivity
conclusion about sha1_git not being an ancestor of.
with_data: indicate to retrieve the content's raw data if path resolves
to a content.
Returns:
The directory pointing to by the revision criterions at path.
"""
rev = lookup_revision_through(revision, limit)
if not rev:
raise NotFoundExc('Revision with criterion %s not found!' % revision)
return (rev['id'],
lookup_directory_with_revision(rev['id'], path, with_data))
def vault_cook(obj_type, obj_id, email=None):
"""Cook a vault bundle.
"""
return vault.cook(obj_type, obj_id, email=email)
def vault_fetch(obj_type, obj_id):
"""Fetch a vault bundle.
"""
return vault.fetch(obj_type, obj_id)
def vault_progress(obj_type, obj_id):
"""Get the current progress of a vault bundle.
"""
return vault.progress(obj_type, obj_id)
def diff_revision(rev_id):
"""Get the list of file changes (insertion / deletion / modification /
renaming) for a particular revision.
"""
rev_sha1_git_bin = _to_sha1_bin(rev_id)
changes = diff.diff_revision(storage, rev_sha1_git_bin,
track_renaming=True)
for change in changes:
change['from'] = converters.from_directory_entry(change['from'])
change['to'] = converters.from_directory_entry(change['to'])
if change['from_path']:
change['from_path'] = change['from_path'].decode('utf-8')
if change['to_path']:
change['to_path'] = change['to_path'].decode('utf-8')
return changes
class _RevisionsWalkerProxy(object):
"""
Proxy class wrapping a revisions walker iterator from
swh-storage and performing needed conversions.
"""
def __init__(self, rev_walker_type, rev_start, *args, **kwargs):
rev_start_bin = hashutil.hash_to_bytes(rev_start)
self.revisions_walker = \
revisions_walker.get_revisions_walker(rev_walker_type,
storage,
rev_start_bin,
*args, **kwargs)
def export_state(self):
return self.revisions_walker.export_state()
def __next__(self):
return converters.from_revision(next(self.revisions_walker))
def __iter__(self):
return self
def get_revisions_walker(rev_walker_type, rev_start, *args, **kwargs):
"""
Utility function to instantiate a revisions walker of a given type,
see :mod:`swh.storage.algos.revisions_walker`.
Args:
rev_walker_type (str): the type of revisions walker to return,
possible values are: ``committer_date``, ``dfs``, ``dfs_post``,
``bfs`` and ``path``
rev_start (str): hexadecimal representation of a revision identifier
args (list): position arguments to pass to the revisions walker
constructor
kwargs (dict): keyword arguments to pass to the revisions walker
constructor
"""
# first check if the provided revision is valid
lookup_revision(rev_start)
return _RevisionsWalkerProxy(rev_walker_type, rev_start, *args, **kwargs)
def lookup_object(object_type: str, object_id: str) -> Dict[str, Any]:
"""
Utility function for looking up an object in the archive by its type
and id.
Args:
object_type (str): the type of object to lookup, either *content*,
*directory*, *release*, *revision* or *snapshot*
object_id (str): the *sha1_git* checksum identifier in hexadecimal
form of the object to lookup
Returns:
Dict[str, Any]: A dictionary describing the object or a list of
dictionary for the directory object type.
Raises:
NotFoundExc: if the object could not be found in the archive
BadInputExc: if the object identifier is invalid
"""
if object_type == CONTENT:
return lookup_content(f'sha1_git:{object_id}')
elif object_type == DIRECTORY:
return {
'id': object_id,
'content': list(lookup_directory(object_id))
}
elif object_type == RELEASE:
return lookup_release(object_id)
elif object_type == REVISION:
return lookup_revision(object_id)
elif object_type == SNAPSHOT:
return lookup_snapshot(object_id)
raise BadInputExc(('Invalid swh object type! Valid types are '
f'{CONTENT}, {DIRECTORY}, {RELEASE} '
f'{REVISION} or {SNAPSHOT}.'))
diff --git a/swh/web/templates/includes/snapshot-context.html b/swh/web/templates/includes/snapshot-context.html
index 49d8ca7c..58b563dc 100644
--- a/swh/web/templates/includes/snapshot-context.html
+++ b/swh/web/templates/includes/snapshot-context.html
@@ -1,31 +1,31 @@
{% comment %}
Copyright (C) 2017-2018 The Software Heritage developers
See the AUTHORS file at the top-level directory of this distribution
License: GNU Affero General Public License version 3, or any later version
See top-level LICENSE file for more information
{% endcomment %}
{% load swh_templatetags %}
\ No newline at end of file
diff --git a/swh/web/templates/includes/top-navigation.html b/swh/web/templates/includes/top-navigation.html
index 5e989b01..1aa12243 100644
--- a/swh/web/templates/includes/top-navigation.html
+++ b/swh/web/templates/includes/top-navigation.html
@@ -1,140 +1,140 @@
{% comment %}
Copyright (C) 2017-2019 The Software Heritage developers
See the AUTHORS file at the top-level directory of this distribution
License: GNU Affero General Public License version 3, or any later version
See top-level LICENSE file for more information
{% endcomment %}
{% load swh_templatetags %}
{% if snapshot_context %}
{% if snapshot_context.branch or snapshot_context.release %}
{% if not snapshot_context or not snapshot_context.is_empty %}
{% include "includes/vault-create-tasks.html" %}
{% endif %}
{% include "includes/show-metadata.html" %}
{% include "includes/take-new-snapshot.html" %}